MCP Builder: Production-Ready Server Development Guide

Table of Contents

1. MCP Fundamentals
2. Tool Design Principles
3. Implementation Patterns
4. Error Handling
5. Testing Strategy
6. Security Best Practices
7. Documentation Standards
8. Production Readiness
9. Common Pitfalls
10. Real-World Examples

MCP Fundamentals

Protocol Architecture

MCP (Model Context Protocol) is a standardized protocol for connecting AI assistants to external data sources and tools. It uses JSON-RPC 2.0 over stdio, HTTP, or SSE transports.

Key Components:

• Server: Exposes tools, resources, and prompts
• Client: Claude Desktop or other MCP clients
• Transport Layer: Communication mechanism (stdio, HTTP, SSE)
• Protocol: JSON-RPC 2.0 messages

Transport Layers

Stdio (Standard Input/Output)

• Best for: Local development, CLI tools
• Pros: Simple, no network setup
• Cons: Single client, no concurrent connections

HTTP

• Best for: Web services, remote servers
• Pros: Scalable, standard protocol
• Cons: Requires HTTP server setup

SSE (Server-Sent Events)

• Best for: Real-time updates, streaming
• Pros: Push updates, efficient
• Cons: One-way from server to client

When to Use MCP vs Alternatives

Use MCP when:

• Building tools for Claude Desktop
• Need standardized tool interface
• Want protocol-level compatibility
• Require resource discovery

Consider alternatives when:

• Building custom integrations
• Need proprietary protocols
• Performance is critical (direct API calls)
• Legacy system integration

Client-Server Communication Patterns

Request-Response Pattern

python
1# FastMCP - Tool execution
2@mcp.tool()
3async def get_user_info(user_id: str) -> dict:
4    """Fetch user information"""
5    return {"id": user_id, "name": "John"}

typescript
1// MCP SDK - Tool execution
2server.setRequestHandler(ListToolsRequestSchema, async () => ({
3  tools: [{
4    name: "get_user_info",
5    description: "Fetch user information",
6    inputSchema: {
7      type: "object",
8      properties: {
9        user_id: { type: "string" }
10      }
11    }
12  }]
13}));

Tool Design Principles

Atomic Operations

✅ GOOD: Atomic, Single Responsibility

python
1# FastMCP
2@mcp.tool()
3async def create_issue(title: str, body: str, repo: str) -> dict:
4    """Create a single GitHub issue"""
5    # One clear action
6    pass
7
8@mcp.tool()
9async def update_issue(issue_id: int, title: str = None, body: str = None) -> dict:
10    """Update an existing GitHub issue"""
11    # Separate update operation
12    pass

typescript
1// MCP SDK
2const createIssue = {
3  name: "create_issue",
4  description: "Create a single GitHub issue in a repository",
5  inputSchema: {
6    type: "object",
7    required: ["title", "body", "repo"],
8    properties: {
9      title: { type: "string", description: "Issue title" },
10      body: { type: "string", description: "Issue body" },
11      repo: { type: "string", description: "Repository name" }
12    }
13  }
14};

❌ BAD: Multi-purpose, Complex

python
1# BAD: Too many responsibilities
2@mcp.tool()
3async def manage_issue(action: str, **kwargs) -> dict:
4    """Create, update, delete, or list issues"""
5    # Too many operations in one tool
6    pass

Clear Input/Output Contracts

✅ GOOD: Explicit Types and Validation

python
1# FastMCP with Pydantic
2from pydantic import BaseModel, Field, validator
3
4class CreateIssueInput(BaseModel):
5    title: str = Field(..., min_length=1, max_length=200)
6    body: str = Field(..., min_length=1)
7    repo: str = Field(..., pattern=r"^[\w\-\.]+/[\w\-\.]+$")
8    labels: list[str] = Field(default_factory=list, max_items=10)
9
10@mcp.tool()
11async def create_issue(input: CreateIssueInput) -> dict:
12    """Create a GitHub issue with validated inputs"""
13    return {
14        "id": 123,
15        "title": input.title,
16        "url": f"https://github.com/{input.repo}/issues/123"
17    }

typescript
1// MCP SDK with Zod validation
2import { z } from "zod";
3
4const CreateIssueSchema = z.object({
5  title: z.string().min(1).max(200),
6  body: z.string().min(1),
7  repo: z.string().regex(/^[\w\-\.]+\/[\w\-\.]+$/),
8  labels: z.array(z.string()).max(10).default([])
9});
10
11server.setRequestHandler(CallToolRequestSchema, async (request) => {
12  if (request.params.name === "create_issue") {
13    const input = CreateIssueSchema.parse(request.params.arguments);
14    // Process validated input
15    return {
16      content: [{
17        type: "text",
18        text: JSON.stringify({ id: 123, title: input.title })
19      }]
20    };
21  }
22});

Parameter Validation

✅ GOOD: Comprehensive Validation

python
1# FastMCP
2from typing import Literal
3from pydantic import BaseModel, validator, HttpUrl
4
5class SearchParams(BaseModel):
6    query: str = Field(..., min_length=1, max_length=500)
7    limit: int = Field(default=10, ge=1, le=100)
8    sort: Literal["relevance", "date", "stars"] = "relevance"
9    
10    @validator("query")
11    def validate_query(cls, v):
12        if any(char in v for char in ["<", ">", "&"]):
13            raise ValueError("Query contains invalid characters")
14        return v.strip()

typescript
1// MCP SDK
2const SearchSchema = z.object({
3  query: z.string().min(1).max(500).refine(
4    (val) => !/[<>&]/.test(val),
5    "Query contains invalid characters"
6  ),
7  limit: z.number().int().min(1).max(100).default(10),
8  sort: z.enum(["relevance", "date", "stars"]).default("relevance")
9});

Descriptive Naming Conventions

✅ GOOD: Clear, Action-Oriented Names

python
1# FastMCP
2@mcp.tool()
3async def fetch_repository_details(owner: str, repo_name: str) -> dict:
4    """Fetch detailed information about a GitHub repository"""
5    pass
6
7@mcp.tool()
8async def list_open_pull_requests(owner: str, repo_name: str) -> list[dict]:
9    """List all open pull requests in a repository"""
10    pass

❌ BAD: Vague Names

python
1# BAD: Unclear purpose
2@mcp.tool()
3async def get_data(owner: str, repo: str) -> dict:
4    """Get data"""
5    pass

When to Split vs Combine Tools

Decision Tree:

1. Different operations? → Split (create vs update vs delete)
2. Different data sources? → Split (GitHub vs GitLab)
3. Different authentication? → Split
4. Same operation, different filters? → Combine with parameters
5. Related operations in same transaction? → Consider combining

✅ GOOD: Appropriate Splitting

python
1# FastMCP - Separate tools for different operations
2@mcp.tool()
3async def create_document(title: str, content: str) -> dict:
4    """Create a new Notion page"""
5    pass
6
7@mcp.tool()
8async def update_document(page_id: str, title: str = None, content: str = None) -> dict:
9    """Update an existing Notion page"""
10    pass
11
12@mcp.tool()
13async def delete_document(page_id: str) -> dict:
14    """Delete a Notion page"""
15    pass

Implementation Patterns

REST API Integration

FastMCP Example:

python
1import httpx
2from fastmcp import FastMCP
3from pydantic import BaseModel, HttpUrl
4
5mcp = FastMCP("GitHub MCP Server")
6
7class GitHubIssue(BaseModel):
8    title: str
9    body: str
10    labels: list[str] = []
11
12@mcp.tool()
13async def create_github_issue(
14    owner: str,
15    repo: str,
16    title: str,
17    body: str,
18    token: str
19) -> dict:
20    """Create a GitHub issue"""
21    async with httpx.AsyncClient() as client:
22        response = await client.post(
23            f"https://api.github.com/repos/{owner}/{repo}/issues",
24            json={"title": title, "body": body},
25            headers={"Authorization": f"token {token}"},
26            timeout=30.0
27        )
28        response.raise_for_status()
29        return response.json()
30
31if __name__ == "__main__":
32    mcp.run()

MCP SDK Example:

typescript
1import { Server } from "@modelcontextprotocol/sdk/server/index.js";
2import { StdioServerTransport } from "@modelcontextprotocol/sdk/server/stdio.js";
3import fetch from "node-fetch";
4
5const server = new Server({
6  name: "github-mcp-server",
7  version: "1.0.0"
8}, {
9  capabilities: {
10    tools: {}
11  }
12});
13
14server.setRequestHandler(CallToolRequestSchema, async (request) => {
15  if (request.params.name === "create_github_issue") {
16    const { owner, repo, title, body, token } = request.params.arguments as any;
17    
18    const response = await fetch(
19      `https://api.github.com/repos/${owner}/${repo}/issues`,
20      {
21        method: "POST",
22        headers: {
23          "Authorization": `token ${token}`,
24          "Content-Type": "application/json"
25        },
26        body: JSON.stringify({ title, body }),
27        signal: AbortSignal.timeout(30000)
28      }
29    );
30    
31    if (!response.ok) {
32      throw new Error(`GitHub API error: ${response.statusText}`);
33    }
34    
35    const data = await response.json();
36    return {
37      content: [{ type: "text", text: JSON.stringify(data) }]
38    };
39  }
40});
41
42async function main() {
43  const transport = new StdioServerTransport();
44  await server.connect(transport);
45}
46
47main().catch(console.error);

Database Connections

FastMCP Example:

python
1import asyncpg
2from fastmcp import FastMCP
3from contextlib import asynccontextmanager
4
5mcp = FastMCP("Database MCP Server")
6pool: asyncpg.Pool = None
7
8@asynccontextmanager
9async def lifespan(app):
10    global pool
11    pool = await asyncpg.create_pool(
12        "postgresql://user:pass@localhost/db",
13        min_size=2,
14        max_size=10
15    )
16    yield
17    await pool.close()
18
19mcp.lifespan = lifespan
20
21@mcp.tool()
22async def query_users(limit: int = 10) -> list[dict]:
23    """Query users from database"""
24    async with pool.acquire() as conn:
25        rows = await conn.fetch(
26            "SELECT id, name, email FROM users LIMIT $1",
27            limit
28        )
29        return [dict(row) for row in rows]

MCP SDK Example:

typescript
1import { Pool } from "pg";
2import { promisify } from "util";
3
4const pool = new Pool({
5  connectionString: "postgresql://user:pass@localhost/db",
6  min: 2,
7  max: 10
8});
9
10server.setRequestHandler(CallToolRequestSchema, async (request) => {
11  if (request.params.name === "query_users") {
12    const { limit = 10 } = request.params.arguments as any;
13    const client = await pool.connect();
14    try {
15      const result = await client.query(
16        "SELECT id, name, email FROM users LIMIT $1",
17        [limit]
18      );
19      return {
20        content: [{
21          type: "text",
22          text: JSON.stringify(result.rows)
23        }]
24      };
25    } finally {
26      client.release();
27    }
28  }
29});

File System Operations

FastMCP Example:

python
1from pathlib import Path
2from fastmcp import FastMCP
3
4mcp = FastMCP("File System MCP Server")
5
6@mcp.tool()
7async def read_file(file_path: str) -> dict:
8    """Read file contents safely"""
9    path = Path(file_path).resolve()
10    
11    # Security: Prevent directory traversal
12    if not str(path).startswith(str(Path.cwd())):
13        raise ValueError("Access denied: path outside workspace")
14    
15    if not path.exists():
16        raise FileNotFoundError(f"File not found: {file_path}")
17    
18    if not path.is_file():
19        raise ValueError(f"Path is not a file: {file_path}")
20    
21    return {
22        "content": path.read_text(),
23        "size": path.stat().st_size
24    }

MCP SDK Example:

typescript
1import * as fs from "fs/promises";
2import * as path from "path";
3
4server.setRequestHandler(CallToolRequestSchema, async (request) => {
5  if (request.params.name === "read_file") {
6    const { file_path } = request.params.arguments as any;
7    const resolved = path.resolve(file_path);
8    const cwd = process.cwd();
9    
10    // Security: Prevent directory traversal
11    if (!resolved.startsWith(cwd)) {
12      throw new Error("Access denied: path outside workspace");
13    }
14    
15    const stats = await fs.stat(resolved);
16    if (!stats.isFile()) {
17      throw new Error("Path is not a file");
18    }
19    
20    const content = await fs.readFile(resolved, "utf-8");
21    return {
22      content: [{
23        type: "text",
24        text: JSON.stringify({ content, size: stats.size })
25      }]
26    };
27  }
28});

Authentication Flows

FastMCP Example:

python
1import os
2from fastmcp import FastMCP
3from typing import Optional
4
5mcp = FastMCP("Authenticated API Server")
6
7class AuthManager:
8    def __init__(self):
9        self._token: Optional[str] = None
10    
11    def get_token(self) -> str:
12        if not self._token:
13            self._token = os.getenv("API_TOKEN")
14            if not self._token:
15                raise ValueError("API_TOKEN environment variable not set")
16        return self._token
17    
18    def refresh_token(self):
19        self._token = None
20
21auth = AuthManager()
22
23@mcp.tool()
24async def make_authenticated_request(endpoint: str) -> dict:
25    """Make authenticated API request"""
26    token = auth.get_token()
27    # Use token in request
28    pass

MCP SDK Example:

typescript
1class AuthManager {
2  private token: string | null = null;
3  
4  getToken(): string {
5    if (!this.token) {
6      this.token = process.env.API_TOKEN || null;
7      if (!this.token) {
8        throw new Error("API_TOKEN environment variable not set");
9      }
10    }
11    return this.token;
12  }
13  
14  refreshToken(): void {
15    this.token = null;
16  }
17}
18
19const auth = new AuthManager();

Proper Async/Await Handling

✅ GOOD: Proper Async Patterns

python
1# FastMCP
2import asyncio
3import httpx
4
5@mcp.tool()
6async def fetch_multiple_resources(urls: list[str]) -> list[dict]:
7    """Fetch multiple resources concurrently"""
8    async with httpx.AsyncClient() as client:
9        tasks = [client.get(url, timeout=10.0) for url in urls]
10        responses = await asyncio.gather(*tasks, return_exceptions=True)
11        
12        results = []
13        for response in responses:
14            if isinstance(response, Exception):
15                results.append({"error": str(response)})
16            else:
17                response.raise_for_status()
18                results.append(response.json())
19        return results

typescript
1// MCP SDK
2async function fetchMultipleResources(urls: string[]): Promise<any[]> {
3  const promises = urls.map(url => 
4    fetch(url, { signal: AbortSignal.timeout(10000) })
5      .then(r => r.ok ? r.json() : { error: r.statusText })
6      .catch(err => ({ error: err.message }))
7  );
8  return Promise.all(promises);
9}

Error Handling

Explicit Error Types

FastMCP Example:

python
1class MCPError(Exception):
2    """Base MCP error"""
3    pass
4
5class ValidationError(MCPError):
6    """Input validation error"""
7    pass
8
9class APIError(MCPError):
10    """External API error"""
11    def __init__(self, message: str, status_code: int = None):
12        super().__init__(message)
13        self.status_code = status_code
14
15class RateLimitError(APIError):
16    """Rate limit exceeded"""
17    pass
18
19@mcp.tool()
20async def api_call(endpoint: str) -> dict:
21    """Make API call with proper error handling"""
22    try:
23        # API call logic
24        pass
25    except httpx.HTTPStatusError as e:
26        if e.response.status_code == 429:
27            raise RateLimitError("Rate limit exceeded", 429)
28        raise APIError(f"API error: {e.response.status_code}", e.response.status_code)
29    except httpx.RequestError as e:
30        raise APIError(f"Request failed: {str(e)}")

MCP SDK Example:

typescript
1class MCPError extends Error {
2  constructor(message: string, public statusCode?: number) {
3    super(message);
4    this.name = "MCPError";
5  }
6}
7
8class ValidationError extends MCPError {}
9class APIError extends MCPError {}
10class RateLimitError extends APIError {}
11
12server.setRequestHandler(CallToolRequestSchema, async (request) => {
13  try {
14    // Tool logic
15  } catch (error: any) {
16    if (error.response?.status === 429) {
17      throw new RateLimitError("Rate limit exceeded", 429);
18    }
19    if (error.response) {
20      throw new APIError(`API error: ${error.response.status}`, error.response.status);
21    }
22    throw new MCPError(`Request failed: ${error.message}`);
23  }
24});

User-Facing Error Messages

✅ GOOD: Clear, Actionable Messages

python
1@mcp.tool()
2async def create_issue(title: str, repo: str) -> dict:
3    """Create GitHub issue"""
4    if not title.strip():
5        raise ValueError(
6            "Issue title cannot be empty. Please provide a descriptive title."
7        )
8    
9    if "/" not in repo:
10        raise ValueError(
11            f"Invalid repository format: '{repo}'. Expected format: 'owner/repo-name'"
12        )

❌ BAD: Technical, Unhelpful

python
1# BAD
2raise ValueError("Invalid input")
3raise Exception("Error occurred")

Retry Logic

FastMCP Example:

python
1import asyncio
2from tenacity import retry, stop_after_attempt, wait_exponential, retry_if_exception_type
3
4@retry(
5    stop=stop_after_attempt(3),
6    wait=wait_exponential(multiplier=1, min=2, max=10),
7    retry=retry_if_exception_type((httpx.RequestError, RateLimitError))
8)
9async def api_call_with_retry(url: str) -> dict:
10    """API call with exponential backoff retry"""
11    async with httpx.AsyncClient() as client:
12        response = await client.get(url, timeout=30.0)
13        if response.status_code == 429:
14            raise RateLimitError("Rate limited")
15        response.raise_for_status()
16        return response.json()

MCP SDK Example:

typescript
1async function retry<T>(
2  fn: () => Promise<T>,
3  maxAttempts: number = 3,
4  delay: number = 1000
5): Promise<T> {
6  for (let i = 0; i < maxAttempts; i++) {
7    try {
8      return await fn();
9    } catch (error) {
10      if (i === maxAttempts - 1) throw error;
11      if (error instanceof RateLimitError) {
12        await new Promise(r => setTimeout(r, delay * Math.pow(2, i)));
13      } else {
14        throw error;
15      }
16    }
17  }
18  throw new Error("Retry failed");
19}

Timeout Handling

FastMCP Example:

python
1import asyncio
2
3@mcp.tool()
4async def long_running_operation(timeout_seconds: int = 30) -> dict:
5    """Operation with timeout"""
6    try:
7        return await asyncio.wait_for(
8            perform_operation(),
9            timeout=timeout_seconds
10        )
11    except asyncio.TimeoutError:
12        raise TimeoutError(
13            f"Operation timed out after {timeout_seconds} seconds. "
14            "Please try again or reduce the scope of the operation."
15        )

MCP SDK Example:

typescript
1async function withTimeout<T>(
2  promise: Promise<T>,
3  timeoutMs: number
4): Promise<T> {
5  return Promise.race([
6    promise,
7    new Promise<T>((_, reject) =>
8      setTimeout(() => reject(new Error(`Timeout after ${timeoutMs}ms`)), timeoutMs)
9    )
10  ]);
11}

Graceful Degradation

FastMCP Example:

python
1@mcp.tool()
2async def fetch_with_fallback(primary_url: str, fallback_url: str) -> dict:
3    """Fetch with fallback on failure"""
4    try:
5        async with httpx.AsyncClient() as client:
6            response = await client.get(primary_url, timeout=10.0)
7            response.raise_for_status()
8            return response.json()
9    except Exception as e:
10        logger.warning(f"Primary source failed: {e}, trying fallback")
11        try:
12            async with httpx.AsyncClient() as client:
13                response = await client.get(fallback_url, timeout=10.0)
14                response.raise_for_status()
15                return response.json()
16        except Exception as fallback_error:
17            raise APIError(
18                f"Both primary and fallback sources failed. "
19                f"Primary: {str(e)}, Fallback: {str(fallback_error)}"
20            )

Logging Strategies

FastMCP Example:

python
1import logging
2import sys
3
4logging.basicConfig(
5    level=logging.INFO,
6    format="%(asctime)s - %(name)s - %(levelname)s - %(message)s",
7    handlers=[
8        logging.FileHandler("mcp-server.log"),
9        logging.StreamHandler(sys.stderr)
10    ]
11)
12
13logger = logging.getLogger("mcp_server")
14
15@mcp.tool()
16async def tool_with_logging(param: str) -> dict:
17    """Tool with structured logging"""
18    logger.info(f"Tool called with param: {param[:50]}")  # Truncate for security
19    try:
20        result = await perform_operation(param)
21        logger.info(f"Tool completed successfully")
22        return result
23    except Exception as e:
24        logger.error(f"Tool failed: {str(e)}", exc_info=True)
25        raise

MCP SDK Example:

typescript
1import * as winston from "winston";
2
3const logger = winston.createLogger({
4  level: "info",
5  format: winston.format.combine(
6    winston.format.timestamp(),
7    winston.format.json()
8  ),
9  transports: [
10    new winston.transports.File({ filename: "mcp-server.log" }),
11    new winston.transports.Console()
12  ]
13});
14
15server.setRequestHandler(CallToolRequestSchema, async (request) => {
16  logger.info("Tool called", { tool: request.params.name });
17  try {
18    // Tool logic
19    logger.info("Tool completed");
20  } catch (error) {
21    logger.error("Tool failed", { error: error.message, stack: error.stack });
22    throw error;
23  }
24});

Testing Strategy

Unit Tests for Individual Tools

FastMCP Example:

python
1import pytest
2from unittest.mock import AsyncMock, patch
3from mcp_server import create_issue
4
5@pytest.mark.asyncio
6async def test_create_issue_success():
7    """Test successful issue creation"""
8    with patch("httpx.AsyncClient") as mock_client:
9        mock_response = AsyncMock()
10        mock_response.json.return_value = {"id": 123, "title": "Test"}
11        mock_response.raise_for_status = AsyncMock()
12        
13        mock_client.return_value.__aenter__.return_value.post.return_value = mock_response
14        
15        result = await create_issue(
16            owner="test",
17            repo="repo",
18            title="Test Issue",
19            body="Body",
20            token="token"
21        )
22        
23        assert result["id"] == 123
24        assert result["title"] == "Test"
25
26@pytest.mark.asyncio
27async def test_create_issue_validation_error():
28    """Test validation error handling"""
29    with pytest.raises(ValueError, match="title cannot be empty"):
30        await create_issue("test", "repo", "", "body", "token")

MCP SDK Example:

typescript
1import { describe, it, expect, vi } from "vitest";
2import { createIssue } from "./tools";
3
4describe("createIssue", () => {
5  it("should create issue successfully", async () => {
6    global.fetch = vi.fn().mockResolvedValue({
7      ok: true,
8      json: async () => ({ id: 123, title: "Test" })
9    });
10    
11    const result = await createIssue({
12      owner: "test",
13      repo: "repo",
14      title: "Test Issue",
15      body: "Body",
16      token: "token"
17    });
18    
19    expect(result.id).toBe(123);
20  });
21  
22  it("should throw validation error for empty title", async () => {
23    await expect(
24      createIssue({ owner: "test", repo: "repo", title: "", body: "body", token: "token" })
25    ).rejects.toThrow("title cannot be empty");
26  });
27});

Integration Tests with Mock Servers

FastMCP Example:

python
1import pytest
2from httpx import ASGITransport, AsyncClient
3from mcp_server import app
4
5@pytest.mark.asyncio
6async def test_integration_create_issue():
7    """Integration test with mock server"""
8    async with AsyncClient(
9        transport=ASGITransport(app=app),
10        base_url="http://test"
11    ) as client:
12        response = await client.post(
13            "/tools/call",
14            json={
15                "name": "create_issue",
16                "arguments": {
17                    "owner": "test",
18                    "repo": "repo",
19                    "title": "Test",
20                    "body": "Body"
21                }
22            }
23        )
24        assert response.status_code == 200
25        data = response.json()
26        assert "id" in data

MCP SDK Example:

typescript
1import { Server } from "@modelcontextprotocol/sdk/server/index.js";
2import { TestTransport } from "./test-transport";
3
4describe("Integration Tests", () => {
5  it("should handle tool call end-to-end", async () => {
6    const transport = new TestTransport();
7    await server.connect(transport);
8    
9    const response = await transport.sendRequest({
10      jsonrpc: "2.0",
11      id: 1,
12      method: "tools/call",
13      params: {
14        name: "create_issue",
15        arguments: {
16          owner: "test",
17          repo: "repo",
18          title: "Test",
19          body: "Body"
20        }
21      }
22    });
23    
24    expect(response.result).toHaveProperty("content");
25  });
26});

Validation of Tool Schemas

FastMCP Example:

python
1import pytest
2from pydantic import ValidationError
3from mcp_server import CreateIssueInput
4
5def test_tool_schema_validation():
6    """Test tool input schema validation"""
7    # Valid input
8    valid = CreateIssueInput(
9        title="Test",
10        body="Body",
11        repo="owner/repo"
12    )
13    assert valid.title == "Test"
14    
15    # Invalid input
16    with pytest.raises(ValidationError):
17        CreateIssueInput(title="", body="Body", repo="owner/repo")
18    
19    # Invalid repo format
20    with pytest.raises(ValidationError):
21        CreateIssueInput(title="Test", body="Body", repo="invalid")

MCP SDK Example:

typescript
1import { describe, it, expect } from "vitest";
2import { CreateIssueSchema } from "./schemas";
3
4describe("Tool Schema Validation", () => {
5  it("should validate correct input", () => {
6    const result = CreateIssueSchema.parse({
7      title: "Test",
8      body: "Body",
9      repo: "owner/repo"
10    });
11    expect(result.title).toBe("Test");
12  });
13  
14  it("should reject invalid input", () => {
15    expect(() => {
16      CreateIssueSchema.parse({ title: "", body: "Body", repo: "owner/repo" });
17    }).toThrow();
18  });
19});

Testing Authentication Flows

FastMCP Example:

python
1@pytest.mark.asyncio
2async def test_authentication_success():
3    """Test successful authentication"""
4    with patch.dict(os.environ, {"API_TOKEN": "test-token"}):
5        auth = AuthManager()
6        token = auth.get_token()
7        assert token == "test-token"
8
9@pytest.mark.asyncio
10async def test_authentication_missing_token():
11    """Test missing token error"""
12    with patch.dict(os.environ, {}, clear=True):
13        auth = AuthManager()
14        with pytest.raises(ValueError, match="API_TOKEN"):
15            auth.get_token()